package ds_industry_2025.ds.ds01.sjwj

import org.apache.spark.sql.functions._
import org.apache.spark.sql.types.DoubleType
import org.apache.spark.sql.{DataFrame, SparkSession}

import java.util.Properties
/*
      剔除订单信息表与订单详细信息表中用户id与商品id不存在现有的维表中的记录，同时建议多利用缓存并充分考虑并行度来优化代码，
      达到更快的计算效果。

1、根据Hive的dwd库中相关表或MySQL中shtd_store中相关表（order_detail、sku_info），计算出与用户id为6708的用户所购买相同
商品种类最多的前10位用户（只考虑他俩购买过多少种相同的商品，不考虑相同的商品买了多少次），将10位用户id进行输出，输出格式如下，将结
果截图粘贴至客户端桌面【Release\任务C提交结果.docx】中对应的任务序号下；
结果格式如下：
-------------------相同种类前10的id结果展示为：--------------------
1,2,901,4,5,21,32,91,14,52
 */
object t1 {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .master("local[*]")
      .appName("t1")
      .config("hive.exec.dynamic.partition.mode","nonstrict")
      .config("spark.serializer","org.apache.spark.serializer.KryoSerializer")
      .config("spark.sql.extensions","org.apache.spark.sql.hudi.HoodieSparkExtension")
      .enableHiveSupport()
      .getOrCreate()

    import spark.implicits._

    val conn=new Properties()
    conn.setProperty("user","root")
    conn.setProperty("password","123456")
    conn.setProperty("driver","com.mysql.jdbc.Driver")

    def read(name:String):DataFrame={
      spark.read.jdbc("jdbc:mysql://192.168.40.110:3306/shtd_store?useSSL=false",name,conn)
    }

    val order_info=read("order_info")
    val order_detail=read("order_detail")
    val user_info=read("user_info")
    val sku_info=read("sku_info")

    order_detail.show

    val users=user_info.select("id").distinct().withColumnRenamed("id","id_1")
    val skus=sku_info.select("id").distinct().withColumnRenamed("id","id_1")

    val order=users.join(order_info,order_info("user_id")===users("id_1"),"inner")
    val detail=skus.join(order_detail,skus("id_1")===order_detail("sku_id"),"inner")

//    val source=order.join(detail)
//      .select("user_id","sku_id")
//      .distinct()

    val source=spark.table("tzgc.source")

    val user_6708_skus = source.filter(col("user_id") === 6708)
      .select("sku_id")
      .map(_(0).toString.toDouble)
      .collect()

    val result = source.filter(col("user_id") !== 6708)
      .withColumn(
        "p",
        when(col("sku_id").cast(DoubleType).isin(user_6708_skus: _*), lit(1.0)).otherwise(0.0)
      )
      .groupBy("user_id")
      .agg(
        sum("p").as("some")
      )
      .orderBy(desc("some"))
      .limit(10)

    println("---------相同种类前10的id结果展示为------------")
    println(result.map(_(0).toString.toInt).collect().mkString(","))





    spark.close()
  }

}
